package com.air.antispider.stream.dataprocess

import com.air.antispider.stream.common.bean._
import com.air.antispider.stream.common.util.jedis.{JedisConnectionUtil, PropertiesUtil}
import com.air.antispider.stream.dataprocess.businessprocess._
import com.air.antispider.stream.dataprocess.constants.BehaviorTypeEnum
import com.air.antispider.stream.dataprocess.constants.TravelTypeEnum.TravelTypeEnum
import kafka.serializer.StringDecoder
import org.apache.commons.lang3.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import redis.clients.jedis.JedisCluster

import scala.collection.mutable.ArrayBuffer


/**
  * 数据预处理的主程序
  */
object DataProcessApp {
  def main(args: Array[String]): Unit = {
    //创建环境
    val conf: SparkConf = new SparkConf().setAppName("DataProcessApp").setMaster("local[*]")
    val sc = new SparkContext(conf)
    //创建SparkContext之后，读取kafka消息之前，先读取mysql中的规则配置，放入广播变量
    val filterRuleList: ArrayBuffer[String] = AnalyzeRuleDB.queryFilterRule()
    //放入广播变量
    //放入广播变量 @volatile 允许多线程安全的更新变量
    @volatile var filterRuleBroadcast: Broadcast[ArrayBuffer[String]] = sc.broadcast(filterRuleList)

    //TODO:将分类规则数据从MySQL中查询出来,放入广播变量
    val classifyRuleMap: Map[String, ArrayBuffer[String]] = AnalyzeRuleDB.queryClassifyRule()
    @volatile var classifyRuleBroadcast: Broadcast[Map[String, ArrayBuffer[String]]] = sc.broadcast(classifyRuleMap)

    //TODO:将解析规则数据从mysql中拿出来，放入广播变量
    val queryAnalyzeRuleList: List[AnalyzeRule] = AnalyzeRuleDB.queryRule(BehaviorTypeEnum.Query.id)
    val bookAnalyzeRuleList: List[AnalyzeRule] = AnalyzeRuleDB.queryRule(BehaviorTypeEnum.Book.id)
    @volatile var queryAnalyzeRuleBroadcast: Broadcast[List[AnalyzeRule]] = sc.broadcast(queryAnalyzeRuleList)
    @volatile var bookAnalyzeRuleBroadcast: Broadcast[List[AnalyzeRule]] = sc.broadcast(bookAnalyzeRuleList)

    //TODO:将黑名单规则数据从mysql中拿出来，判断是否为高频ip，放入广播变量
    val blackIPList: ArrayBuffer[String] = AnalyzeRuleDB.queryBlackIPList()
    @volatile var blackIPListBroadcast: Broadcast[ArrayBuffer[String]] = sc.broadcast(blackIPList)


    val ssc = new StreamingContext(sc, Seconds(2))
    //设置日志级别
    ssc.sparkContext.setLogLevel("WARN")
    //TODO:从kafka中获取消息
    /**
      * K: ClassTag,
      * V: ClassTag,
      * KD <: Decoder[K]: ClassTag,
      * VD <: Decoder[V]: ClassTag] (
      * ssc: StreamingContext,
      * kafkaParams: Map[String, String],
      * topics: Set[String]
      */
    //使用工具类PropertiesUtil，获取kafka的参数
    val brokerList: String = PropertiesUtil.getStringByKey("default.brokers", "kafkaConfig.properties")
    val topic: String = PropertiesUtil.getStringByKey("source.nginx.topic", "kafkaConfig.properties")
    //定义kafka的连接参数
    val kafkaParams: Map[String, String] = Map[String, String](
      "metadata.broker.list" -> brokerList)
    //定义topics信息
    val topics: Set[String] = Set[String](topic)
    //从Kafka中获取消息, StringDecoder因为Kafka使用java默认的序列化效率不高,
    // 所以Kafka自己捣鼓了一套序列化机制,这里就使用Kafka的序列化
    val sourceStream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
    //打印数据
    //直接输出数据是一个map集合，第一个字段是key(就是分区值),第二个是value(消息)
    //sourceStream.print()
    //取出原始的数据
    val sourceMessage: DStream[String] = sourceStream.map(_._2)
    //TODO:获取redis的连接
    val jedis: JedisCluster = JedisConnectionUtil.getJedisCluster
    sourceMessage.foreachRDD(messageRDD => {
      //TODO:操作数据之前，先更新redis广播变量
      var filterChangeFlag: String = jedis.get("FilterChangeFlag")
      //进行判断
      if (StringUtils.isEmpty(filterChangeFlag)) {
        //如果取出数据为空，就先创建
        filterChangeFlag = "true"
        //将数据放到广播变量中
        jedis.set("FilterChangeFlag", filterChangeFlag)
      }
      //如果不为空
      if (filterChangeFlag.toBoolean) {
        //更新广播变量
        //先废除之前的广播变量
        filterRuleBroadcast.unpersist()
        //查询最新数据
        val filterRuleList: ArrayBuffer[String] = AnalyzeRuleDB.queryFilterRule()
        //放入广播变量
        filterRuleBroadcast = sc.broadcast(filterRuleList)
        //更新redis
        filterChangeFlag = "false"
        jedis.set("FilterChangeFlag", filterChangeFlag)
      }

      //TODO:从Redis获取规则状态信息,看是否发生修改
      var classifyRuleChangeFlag: String = jedis.get("ClassifyRuleChangeFlag")
      if (StringUtils.isBlank(classifyRuleChangeFlag)) {
        //如果没有初始值,那么就给个true
        classifyRuleChangeFlag = "true"
        jedis.set("ClassifyRuleChangeFlag", classifyRuleChangeFlag)
      }
      if (classifyRuleChangeFlag.toBoolean) {
        //说明需要更新广播变量
        //先废除之前的广播变量
        classifyRuleBroadcast.unpersist()
        //查询最新的数据
        val classifyRuleMap: Map[String, ArrayBuffer[String]] = AnalyzeRuleDB.queryClassifyRule()
        //更新广播变量
        classifyRuleBroadcast = sc.broadcast(classifyRuleMap)
        //更新redis信息
        classifyRuleChangeFlag = "false"
        jedis.set("ClassifyRuleChangeFlag", classifyRuleChangeFlag)
      }

      //TODO:从Redis获取规则状态信息,看是否发生修改
      var analyzeRuleChangeFlag: String = jedis.get("AnalyzeRuleChangeFlag")
      if (StringUtils.isBlank(analyzeRuleChangeFlag)) {
        //如果没有初始值,那么就给个true
        analyzeRuleChangeFlag = "true"
        jedis.set("AnalyzeRuleChangeFlag", analyzeRuleChangeFlag)
      }
      if (analyzeRuleChangeFlag.toBoolean) {
        //说明需要更新广播变量
        //先废除之前的广播变量
        queryAnalyzeRuleBroadcast.unpersist()
        bookAnalyzeRuleBroadcast.unpersist()
        //查询最新的数据
        val queryAnalyzeRuleList: List[AnalyzeRule] = AnalyzeRuleDB.queryRule(BehaviorTypeEnum.Query.id)
        val bookAnalyzeRuleList: List[AnalyzeRule] = AnalyzeRuleDB.queryRule(BehaviorTypeEnum.Book.id)
        //更新广播变量
        queryAnalyzeRuleBroadcast = sc.broadcast(queryAnalyzeRuleList)
        bookAnalyzeRuleBroadcast = sc.broadcast(bookAnalyzeRuleList)
        //更新redis信息
        analyzeRuleChangeFlag = "false"
        jedis.set("AnalyzeRuleChangeFlag", analyzeRuleChangeFlag)
      }

      //TODO: 从Redis获取黑名单规则状态信息,看是否发生修改，判断是否为高频ip
      var blackListChangeFlag: String = jedis.get("blackListChangeFlag")
      //如果为空，就设置默认值
      if (StringUtils.isEmpty(blackListChangeFlag)) {
        blackListChangeFlag = "true"
        jedis.set("blackListChangeFlag", blackListChangeFlag)
      }
      //如果规则改变了
      if (blackListChangeFlag.toBoolean) {
        //先废除以前的广播变量
        blackIPListBroadcast.unpersist()
        //从数据库中更新数据
        //更新广播变量
        val blackIPList: ArrayBuffer[String] = AnalyzeRuleDB.queryBlackIPList()
        blackIPListBroadcast = sc.broadcast(blackIPList)
        //更新redis的信息
        blackListChangeFlag = "false"
        //重新设置redis信息
        jedis.set("blackListChangeFlag", blackListChangeFlag)
      }



      //TODO:链路统计功能
      BusinessProcess.linkCount(messageRDD)

      //TODO:进行数据过滤的操作
      val filterRDD: RDD[String] = messageRDD.filter(message => FilterURL.filterRule(message, filterRuleBroadcast.value))

      //TODO:对用户敏感信息进行加密操作
      //加密手机号
      val encryptedPhoneRDD: RDD[String] = EncryptedData.encryptedPhoneNum(filterRDD)
      //加密身份证号
      val encryptedRDD: RDD[String] = EncryptedData.encryptedIDNum(encryptedPhoneRDD)

      //TODO:最终需要的是一个结构化数据String => processedData结构化的数据
      val processedDataRDD: RDD[ProcessedData] = encryptedRDD.map(message => {
        //数据切割
        val (request,
        requestMethod,
        contentType,
        requestBody,
        httpReferrer,
        remoteAddr,
        httpUserAgent,
        timeIso8601,
        serverAddr,
        cookiesStr,
        cookieValue_JSESSIONID,
        cookieValue_USERID) = DataSplit.dataSplit(message)

        //TODO:对数据打标签，看看当前请求是国内查询/预定，还是国际查询/预定
        val requestType: RequestType = RequestTypeClassifier.classifyByRequest(request, classifyRuleBroadcast.value)

        //TODO:往返信息打标签
        val travelType: TravelTypeEnum = TravelTypeClassifier.travelTypeClassifier(httpReferrer)

        //TODO:将请求体数据，请求url，请求方式，标签信息，解析规则放入工具类中开始解析
        //开始解析查询相关参数
        val queryRequestData: Option[QueryRequestData] = AnalyzeRequest.analyzeQueryRequest(requestType, requestMethod, contentType, request, requestBody, travelType, queryAnalyzeRuleBroadcast.value)
        //开始解析预定相关参数
        val bookRequestData: Option[BookRequestData] = AnalyzeBookRequest.analyzeBookRequest(requestType, requestMethod, contentType, request, requestBody, travelType, bookAnalyzeRuleBroadcast.value)

        //TODO:数据加工，判断是否为高频ip。
        val isFreIP: Boolean = IpOperation.isFreIP(remoteAddr, blackIPListBroadcast.value)

        //TODO: //数据结构化
        val processedData: ProcessedData = DataPackage.dataPackage("",
          requestMethod,
          request,
          remoteAddr,
          httpUserAgent,
          timeIso8601,
          serverAddr,
          isFreIP,
          requestType,
          travelType,
          cookieValue_JSESSIONID,
          cookieValue_USERID,
          queryRequestData, //查询数据的请求参数封装
          bookRequestData, //预定数据的请求参数封装
          httpReferrer)
        processedData
      })
      //TODO:将数据发送到kafka中
      DataSend.sendDataToKafka(processedDataRDD)
      processedDataRDD.foreach(println)
      //encryptedRDD.foreach(println)
    })
    //启动程序
    ssc.start()
    //让程序一直运行
    ssc.awaitTermination()

  }
}
